package etcd

import (
	"context"
	"encoding/json"
	"fmt"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.uber.org/zap"
	"go_kafka_log/common"
	"go_kafka_log/tailf"
	"time"
)

/*
mac 启动 etcd
https://etcd.io/docs/v3.5/quickstart/

./etcdctl put name 999
./etcdctl get name
./etcdctl del name
./etcdctl watch name
*/

var (
	client *clientv3.Client
)

func Init(addr []string) (err error) {

	client, err = clientv3.New(clientv3.Config{
		Endpoints:   addr,
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		// handle error!
		fmt.Printf("connect to etcd failed, err:%v\n", err)
		return
	}
	fmt.Println("connect to etcd success")

	return
}

// 拉取日志收集配置项
func GetConf(key string) (collectEntryList []common.CollectEntry, err error) {

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*2)
	defer func() {
		cancel()

	}()
	resp, err := client.Get(ctx, key)
	if err != nil {
		err = fmt.Errorf("etcd 不存在对应的 key: %s, err: %v", key, err)

		return
	}

	if len(resp.Kvs) == 0 {
		err = fmt.Errorf("etcd 对应 Key：%s，无数据", key)
		return
	}

	ret := resp.Kvs[0]
	err = json.Unmarshal(ret.Value, &collectEntryList)
	if err != nil {
		err = fmt.Errorf("序列化对应 etcd key: %s 失败, err: %v", key, err)
		return
	}
	return
}

// 监控日志收集项
func WatchConf(key string) {
	var newConf []common.CollectEntry

	wCh := client.Watch(context.Background(), key)
	for wresp := range wCh {

		for _, ev := range wresp.Events {
			fmt.Printf("Type: %s Key:%s Value:%s\n", ev.Type, ev.Kv.Key, ev.Kv.Value)

			// 如果是删除操作
			if ev.Type != clientv3.EventTypeDelete {

				err := json.Unmarshal(ev.Kv.Value, &newConf)
				if err != nil {

					zap.L().Error("etcd watch 新配置项时 json 反序列化失败， err is", zap.Error(err))
					continue
				}

			}

			// 告诉 tailf 包使用新的配置
			tailf.SendNewConf(newConf)
		}
	}

}
